Action Operators — Cloud Provider Integration

Cloud Provider Operators

AWS Example: S3 to Redshift Pipeline

from airflow.sdk import DAG
from airflow.providers.amazon.aws.operators.s3 import S3CreateObjectOperator
from airflow.providers.amazon.aws.transfers.s3_to_redshift import S3ToRedshiftOperator
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from datetime import datetime

with DAG("s3_to_redshift", schedule="@daily", start_date=datetime(2024, 1, 1)) as dag:

    # Wait for source file to arrive
    wait_for_file = S3KeySensor(
        task_id="wait_for_csv",
        bucket_name="raw-data-bucket",
        bucket_key="sales/{{ ds }}/sales_data.csv",
        aws_conn_id="aws_default",
        deferrable=True,           # Don't block a worker slot
    )

    # Copy from S3 to Redshift
    load_to_redshift = S3ToRedshiftOperator(
        task_id="load_to_redshift",
        schema="public",
        table="sales_staging",
        s3_bucket="raw-data-bucket",
        s3_key="sales/{{ ds }}/sales_data.csv",
        redshift_conn_id="redshift_default",
        aws_conn_id="aws_default",
        copy_options=["CSV", "IGNOREHEADER 1", "REGION 'us-east-1'"],
    )

    wait_for_file >> load_to_redshift

GCP Example: BigQuery Pipeline

from airflow.sdk import DAG
from airflow.providers.google.cloud.operators.bigquery import (
    BigQueryCreateEmptyDatasetOperator,
    BigQueryInsertJobOperator,
)
from datetime import datetime

with DAG("bigquery_pipeline", schedule="@daily", start_date=datetime(2024, 1, 1)) as dag:

    create_dataset = BigQueryCreateEmptyDatasetOperator(
        task_id="create_dataset",
        dataset_id="analytics",
        gcp_conn_id="google_cloud_default",
    )

    run_query = BigQueryInsertJobOperator(
        task_id="aggregate_sales",
        configuration={
            "query": {
                "query": """
                    SELECT
                        DATE(order_date) as order_day,
                        SUM(amount) as total_sales,
                        COUNT(*) as order_count
                    FROM `project.raw.orders`
                    WHERE DATE(order_date) = '{{ ds }}'
                    GROUP BY 1
                """,
                "useLegacySql": False,
                "destinationTable": {
                    "projectId": "my-project",
                    "datasetId": "analytics",
                    "tableId": "daily_sales_{{ ds_nodash }}",
                },
                "writeDisposition": "WRITE_TRUNCATE",
            }
        },
        gcp_conn_id="google_cloud_default",
    )

    create_dataset >> run_query

This content is reserved for Premium Members.